package com.abyss.sink.stream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * 将数据写出到Redis中
 */
public class RedisSinkDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. Source
        // 将数据转换为Tuple2, 注意, socket 传入的数据需要是kv字符串以空格分隔
        DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);

        // 3. 将数据转变为Tuple2

        SingleOutputStreamOperator<Tuple2<String, String>> map = socketTextStream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                return Tuple2.of(value.split(" ")[0], value.split(" ")[1]);
            }
        });

        // 4. 将数据写出到Redis中
        FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder().setHost("node1").setPort(6379).build();
        map.addSink(new RedisSink<>(redisConfig, new MyRedisMapper()));

        // 5. 执行
        env.execute();
    }
    // 基于自定义的RedisMapper实现来构建RedisSink
    public static class MyRedisMapper implements RedisMapper<Tuple2<String, String>> {
        /*
      描述要写出的数据类型
       */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.LPUSH);
        }
        // 设置key
        @Override
        public String getKeyFromData(Tuple2<String, String> data) {
            return data.f0;
        }
        // 设置value
        @Override
        public String getValueFromData(Tuple2<String, String> data) {
            return data.f1;
        }
    }
}